package com.atguigu.tingshu.search.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.extra.pinyin.PinyinUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.Buckets;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.cloud.commons.lang.StringUtils;
import com.atguigu.tingshu.album.AlbumFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.function.context.test.FunctionalSpringBootTest;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private AlbumFeignClient albumFeignClient;

    @Autowired
    private UserFeignClient userFeignClient;

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    private static final String INDEX_NAME = "albuminfo";
    private static final String SUGGEST_INDEX_NAME = "suggestinfo";


    /**
     * 上架专辑-导入索引库
     * @param albumId
     */
    @Override
    public void upperAlbum(Long albumId) {

        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            //根据商品id查询商品信息，需要调用其他微服务
            AlbumInfo albumInfo = albumFeignClient.getAlbumInfo(albumId).getData();
            if (albumInfo == null) {
                throw new RuntimeException("专辑不存在");
            }
            List<AttributeValueIndex> attributeValueIndexList = albumInfo.getAlbumAttributeValueVoList()
                    .stream()
                    .map(albumAttributeValue -> {
                        return BeanUtil.copyProperties(albumAttributeValue, AttributeValueIndex.class);
                    }).collect(Collectors.toList());
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
            //上传专辑信息
            BeanUtil.copyProperties(albumInfo, albumInfoIndex);
            return albumInfo;
        }, threadPoolExecutor);

        CompletableFuture<Void> userInfoCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //查询用户表
            UserInfoVo userInfoVo = userFeignClient.getUserInfoVo(albumInfo.getUserId()).getData();
            if (userInfoVo == null) {
                throw new RuntimeException("用户不存在");
            }
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
            //设置上架时间
            albumInfoIndex.setCreateTime(new Date());
        }, threadPoolExecutor);

        CompletableFuture<Void> baseCategoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //查询三级分类
            BaseCategoryView baseCategoryView = albumFeignClient.getCategoryView(albumInfo.getCategory3Id()).getData();
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
        }, threadPoolExecutor);

        CompletableFuture<Void> statCompletableFuture = CompletableFuture.runAsync(() -> {
            //随机生成统计数量
            // 生成随机数
            int playStatNum = (int) (Math.random() * 1000);
            int subscribeStatNum = (int) (Math.random() * 1000);
            int buyStatNum = (int) (Math.random() * 1000);
            int commentStatNum = (int) (Math.random() * 1000);

            // 设置统计数值
            albumInfoIndex.setPlayStatNum(playStatNum);
            albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
            albumInfoIndex.setBuyStatNum(buyStatNum);
            albumInfoIndex.setCommentStatNum(commentStatNum);

            // 随机生成评分
            albumInfoIndex.setHotScore(playStatNum * 0.6 + subscribeStatNum * 0.4 + buyStatNum * 0.8 + commentStatNum * 0.2);
        }, threadPoolExecutor);

        CompletableFuture.allOf(albumInfoCompletableFuture, userInfoCompletableFuture, baseCategoryCompletableFuture, statCompletableFuture).join();

        albumInfoIndexRepository.save(albumInfoIndex);
        this.saveSuggestIndex(albumInfoIndex);


        //将新增的商品SKUID存入布隆过滤器
        //获取布隆过滤器，将新增skuID存入布隆过滤器
        RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumId);
    }

    /**
     * 下架专辑-删除索引库
     * @param albumId
     */
    @Override
    public void lowerAlbum(Long albumId) {
        albumInfoIndexRepository.deleteById(albumId);
    }

    /**
     * 专辑检索
     * @param jsonString
     * @return
     */
    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        try {
            //创建DSL
            SearchRequest searchRequest = this.buildDSL(albumIndexQuery);
            System.out.println("searchRequest = " + searchRequest);

            //检索dsl
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);

            //解析响应结果
            return this.parseResult(searchResponse, albumIndexQuery);
        } catch (Exception e) {
            log.error("[搜索服务]查询条件：{}，站内检索异常：{}", albumIndexQuery, e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 查询指定一级分类下热门排行专辑
     * @param category1Id 一级分类ID
     * @return List<Map<String, Object>> 返回包含三级分类信息和对应热门专辑列表的Map集合
     *         Map中包含两个键值对：
     *         - baseCategory3: BaseCategory3对象，表示三级分类信息
     *         - list: List<AlbumInfoIndex>对象，表示该三级分类下的热门专辑列表
     */
    @Override
    public List<Map<String, Object>> getTopCategory3HotAlbumList(Long category1Id) {
        // 获取指定一级分类下的前7个热门三级分类
        List<BaseCategory3> baseCategory3List = albumFeignClient.getTop7BaseCategory3(category1Id).getData();
        Assert.notNull(baseCategory3List, "一级分类{}未包含置顶三级分类", category1Id);
        
        // 提取三级分类ID列表并创建ID到分类的映射关系
        List<Long> baseCategory3IdList = baseCategory3List.stream().map(BaseCategory3::getId).collect(Collectors.toList());
        Map<Long, BaseCategory3> category3Map = baseCategory3List.stream().collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        
        // 创建Elasticsearch查询所需的FieldValue列表
        List<FieldValue> fieldValueList = baseCategory3IdList.stream()
                .map(baseCategory3Id -> FieldValue.of(baseCategory3Id))
                .collect(Collectors.toList());
        
        try {
            // 构建Elasticsearch查询：
            // 1. 根据三级分类ID过滤数据
            // 2. 聚合统计每个三级分类的专辑信息
            // 3. 在每个三级分类中获取前6个最热门的专辑
            SearchResponse<AlbumInfoIndex> searchResponse = elasticsearchClient.search(
                    s -> s.index(INDEX_NAME).size(0)
                            .query(q -> q.terms(t -> t.field("category3Id").terms(tf -> tf.value(fieldValueList))))
                            .aggregations("category3Agg", a -> a.terms(t -> t.field("category3Id").size(7)
                            ).aggregations("top6Agg", a1 -> a1.topHits(t -> t.size(6).sort(sort -> sort.field(f -> f.field("hotScore").order(SortOrder.Desc)))))),
                    AlbumInfoIndex.class);
            System.out.println(searchResponse);
            Aggregate category3Agg = searchResponse.aggregations().get("category3Agg");
            Buckets<LongTermsBucket> buckets = category3Agg.lterms().buckets();
            List<LongTermsBucket> bucketList = buckets.array();
            if (CollectionUtil.isNotEmpty(bucketList)){
                // 处理聚合结果，将每个三级分类及其对应的热门专辑封装成Map结构返回
                List<Map<String, Object>> listMap = bucketList.stream().map(bucket -> {
                    Map<String, Object> map = new HashMap<>();
                    long category3Id = bucket.key();
                    BaseCategory3 baseCategory3 = category3Map.get(category3Id);
                    map.put("baseCategory3", baseCategory3);
                    Aggregate top6Agg = bucket.aggregations().get("top6Agg");
                    List<Hit<JsonData>> hits = top6Agg.topHits().hits().hits();
                    if (CollectionUtil.isNotEmpty(hits)){
                        // 提取专辑信息并转换为AlbumInfoIndex对象
                        List<AlbumInfoIndex> hotAlbumList = hits.stream().map(hit -> {
                            JsonData source = hit.source();
                            return source.to(AlbumInfoIndex.class);
                        }).collect(Collectors.toList());
                        map.put("list", hotAlbumList);
                    }
                    return map;
                }).collect(Collectors.toList());
                return listMap;
            }
        } catch (Exception e) {
            log.error("[检索服务]首页热门专辑异常：{}", e);
            throw new RuntimeException(e);
        }
        return null;
    }

    /**
     * 上传搜索索引
     * @param albumInfoIndex
     */
    @Override
    public void saveSuggestIndex(AlbumInfoIndex albumInfoIndex) {
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        String albumTitle = albumInfoIndex.getAlbumTitle();
        suggestIndex.setTitle(albumTitle);
        suggestIndex.setKeyword(new Completion(new String[]{albumTitle}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinyinUtil.getPinyin(albumTitle, "")}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinyinUtil.getFirstLetter(albumTitle, "")}));
        suggestIndexRepository.save(suggestIndex);
    }

    /**
     * 关键字自动补全
     * @param keyword
     * @return
     */
    @Override
    @SneakyThrows
    public List<String> completeSuggest(String keyword) {
        // es查询
        SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(
                s -> s.index(SUGGEST_INDEX_NAME)
                        .suggest(su -> su
                                .suggesters("mySuggestKeyword", s1 -> s1
                                        .prefix(keyword).completion(c -> c.field("keyword").skipDuplicates(true).size(10))
                                )
                                .suggesters("mySuggestPinyin", s1 -> s1
                                        .prefix(keyword).completion(c -> c.field("keywordPinyin").skipDuplicates(true))
                                )
                                .suggesters("mySuggestSequence", s1 -> s1
                                        .prefix(keyword).completion(c -> c.field("keywordSequence").skipDuplicates(true))
                                )
                        )
                , SuggestIndex.class);
        //解析结果
        LinkedHashSet<String> hashSet = new LinkedHashSet<>();
        List<Suggestion<SuggestIndex>> mySuggestKeyword = searchResponse.suggest().get("mySuggestKeyword");
        List<Suggestion<SuggestIndex>> mySuggestPinyin = searchResponse.suggest().get("mySuggestPinyin");
        List<Suggestion<SuggestIndex>> mySuggestSequence = searchResponse.suggest().get("mySuggestSequence");
        hashSet.addAll(this.parseSuggestResult(mySuggestKeyword));
        hashSet.addAll(this.parseSuggestResult(mySuggestPinyin));
        hashSet.addAll(this.parseSuggestResult(mySuggestSequence));
        if (hashSet.size() >= 10){
            return new ArrayList<>(hashSet).subList(0, 10);
        }
        //查询ablum
        SearchResponse<AlbumInfoIndex> allbumSearchResponse = elasticsearchClient.search(s -> s
                .index(INDEX_NAME).query(q -> q.match(m -> m.field("albumTitle").query(keyword)))
                , AlbumInfoIndex.class);
        //解析结果
        List<String> albumInfoIndexList = allbumSearchResponse.hits().hits().stream().map(hit -> hit.source().getAlbumTitle()).collect(Collectors.toList());
        hashSet.addAll(albumInfoIndexList);
        if (hashSet.size() >= 10){
            return new ArrayList<>(hashSet).subList(0, 10);
        }
        //输出结果
        return new ArrayList<>(hashSet);
    }

    /**
     * 查询专辑详情
     * @param albumId
     * @return
     */
    @Override
    public Map<String, Object> getAlbumInfo(Long albumId) {
        //0.查询布隆过滤器是否包含查询专辑ID
        RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        boolean flag = bloomFilter.contains(albumId);
        if (!flag) {
            throw new GuiguException(404, "访问专辑不存在");
        }
        HashMap<String, Object> stringObjectHashMap = new HashMap<>();
        //专辑信息
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            Result<AlbumInfo> albumInfo = albumFeignClient.getAlbumInfo(albumId);
            stringObjectHashMap.put("albumInfo", albumInfo.getData());
            return albumInfo.getData();
        }, threadPoolExecutor);

        //获取主播信息
        CompletableFuture<Void> announcerCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfoVo> userInfoVo = userFeignClient.getUserInfoVo(albumInfo.getUserId());
            stringObjectHashMap.put("announcer", userInfoVo.getData());
        }, threadPoolExecutor);

        //专辑分类
        CompletableFuture<Void> baseCategoryViewCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> categoryView = albumFeignClient.getCategoryView(albumInfo.getCategory3Id());
            stringObjectHashMap.put("baseCategoryView", categoryView.getData());
        }, threadPoolExecutor);

        //专辑统计信息
        CompletableFuture<Void> albumStatVoCompletableFuture = CompletableFuture.runAsync(() -> {
            Result<AlbumStatVo> albumStatVo = albumFeignClient.getAlbumStatVo(albumId);
            stringObjectHashMap.put("albumStatVo", albumStatVo.getData());
        }, threadPoolExecutor);

        CompletableFuture.allOf(albumInfoCompletableFuture, announcerCompletableFuture, baseCategoryViewCompletableFuture, albumStatVoCompletableFuture).join();

        return stringObjectHashMap;
    }

    @Override
    public List<AlbumInfoIndex> findRankingList(Long category1Id, String dimension) {
        String key = RedisConstant.RANKING_KEY_PREFIX + category1Id;
        Boolean b = redisTemplate.opsForHash().hasKey(key, dimension);
        if ( b){
            return (List<AlbumInfoIndex>) redisTemplate.opsForHash().get(key, dimension);
        }
        return null;

    }

    /**
     * 解析搜索结果
     * @param mySuggestKeyword
     */
    private Collection<String> parseSuggestResult(List<Suggestion<SuggestIndex>> mySuggestKeyword) {
        ArrayList<String> list = new ArrayList<>();
        mySuggestKeyword.forEach(suggestion -> {
            suggestion.completion().options().forEach(option -> {
                list.add(option.source().getTitle());
            });
        });
        return list;
    }


    /**
     * 构建DSL
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest buildDSL(AlbumIndexQuery albumIndexQuery) {
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index(INDEX_NAME);
        BoolQuery.Builder allBoolQueryBuilder = new BoolQuery.Builder();
        String keyword = albumIndexQuery.getKeyword();
        if (StringUtils.isNotBlank( keyword)) {
            BoolQuery.Builder keyWordBoolQueryBuilder = new BoolQuery.Builder();
            //匹配专辑名
            keyWordBoolQueryBuilder.should(s->s.match(m->m.field("albumTitle").query(keyword)));
            //匹配专辑简介
            keyWordBoolQueryBuilder.should(s->s.match(m->m.field("albumIntro").query(keyword)));
            //匹配主播名
            keyWordBoolQueryBuilder.should(s->s.term(t->t.field("announcerName").value(keyword)));
            allBoolQueryBuilder.must(keyWordBoolQueryBuilder.build()._toQuery());
        }

        //查询分类
        if (albumIndexQuery.getCategory1Id() != null){
            allBoolQueryBuilder.filter(f->f.term(t->t.field("category1Id").value(albumIndexQuery.getCategory1Id())));
        }
        if (albumIndexQuery.getCategory2Id() != null){
            allBoolQueryBuilder.filter(f->f.term(t->t.field("category2Id").value(albumIndexQuery.getCategory2Id())));
        }
        if (albumIndexQuery.getCategory3Id() != null){
            allBoolQueryBuilder.filter(f->f.term(t->t.field("category3Id").value(albumIndexQuery.getCategory3Id())));
        }

        //查询属性
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (CollectionUtil.isNotEmpty(attributeList)){
            for (String attribute : attributeList){
                String[] split = attribute.split(":");
                if (split != null  && split.length == 2){
                    allBoolQueryBuilder.filter(f->f
                            .nested(n->n
                                    .path("attributeValueIndexList")
                                    .query(q->q
                                            .bool(b->b
                                                    .must(m->m.term(t->t.field("attributeValueIndexList.attributeId").value(split[0])))
                                                    .must(m->m.term(t->t.field("attributeValueIndexList.valueId").value(split[1])))
                                            )
                                    )
                            )
                    );
                }
            }
        }

        //总和查询条件
        builder.query(allBoolQueryBuilder.build()._toQuery());

        //分页
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        builder.from((pageNo - 1) * pageSize);
        builder.size(pageSize);

        //排序
        String order = albumIndexQuery.getOrder();
        if (StringUtils.isNotBlank(order)){
            String[] split = order.split(":");
            if (split.length == 2 && split != null){
                String orderField = "";
                switch (split[0]){
                    case "1":
                        orderField = "hotScore";
                        break;
                    case "2":
                        orderField = "playStatNum";
                        break;
                    case "3":
                        orderField = "createTime";
                        break;
                }
                String finalOrderField = orderField;
                builder.sort(s -> s.field(f -> f.field(finalOrderField).order("asc".equals(split[1]) ? SortOrder.Asc : SortOrder.Desc)));
            }
        }


        //高亮
        if (StringUtils.isNotBlank(keyword)){
            builder.highlight(h->h.fields("albumTitle",hf -> hf.preTags("<font style='color:red'>").postTags("</font>")));
        }


        //设置请求体参数"_source" 处理字段指定
        builder.source(s -> s
                .filter(f -> f
                        .excludes("category1Id",
                                "category2Id",
                                        "category3Id",
                                        "attributeValueIndexList.attributeId",
                                        "attributeValueIndexList.valueId")));

        return builder.build();
    }

    /**
     * 解析es结果
     * @param searchResponse
     * @return
     */
    private AlbumSearchResponseVo parseResult(SearchResponse<AlbumInfoIndex> searchResponse, AlbumIndexQuery albumIndexQuery) {
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        //获取分页信息
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        albumSearchResponseVo.setPageNo(pageNo);
        albumSearchResponseVo.setPageSize(pageSize);
        albumSearchResponseVo.setTotal(searchResponse.hits().total().value());
        Long total = albumSearchResponseVo.getTotal();
        // 计算总页数并设置到响应对象中
        Long totalPages = total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        albumSearchResponseVo.setTotalPages(totalPages);

        List<Hit<AlbumInfoIndex>> hitList = searchResponse.hits().hits();
        if (CollectionUtil.isNotEmpty(hitList)){
            List<AlbumInfoIndexVo> albumInfoIndexVoList = hitList.stream().map(hit -> {
                //拷贝查询的信息
                AlbumInfoIndexVo albumInfoIndexVo = BeanUtil.copyProperties(hit.source(), AlbumInfoIndexVo.class);
                //高亮
                Map<String, List<String>> highlightMap = hit.highlight();
                if (CollectionUtil.isNotEmpty(highlightMap) && highlightMap.containsKey("albumTitle")) {
                    String highlightAlbumTitle = highlightMap.get("albumTitle").get(0);
                    albumInfoIndexVo.setAlbumTitle(highlightAlbumTitle);
                }
                return albumInfoIndexVo;
            }).collect(Collectors.toList());
            albumSearchResponseVo.setList(albumInfoIndexVoList);
        }
        return albumSearchResponseVo;
    }

}
